package org.example.rocketmq.msg_type.filter;

import org.apache.commons.lang3.ThreadUtils;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.MessageQueueSelector;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageQueue;
import org.apache.rocketmq.remoting.common.RemotingHelper;

import java.time.Duration;
import java.util.List;

/**
 * 过滤消息
 * RocketMQ 的最佳实践中就建议使用 RocketMQ 时，一个应用可以就用一个 Topic，而应用中的不同业务就用 Tag 来区分。
 * Tag 方式有一个很大的限制，就是一个消息只能有一个 Tag，这在一些比较复杂的场景就有点不足了 这时候可以使用 SQL 表达式来对消息进行过滤。
 */
public class ProducerFilterSql {

    public static void main(String[] args) throws Exception{
        producer();
    }


    /**
     * 顺序消息值生产者局部有序发送到一个 queue, 但多个 queue 之间是全局无序的。
     * 顺序消息生产者样例：通过 MessageQueueSelector将消息有序发送到同一个queue中
     * @throws Exception
     */
    public static void producer() throws Exception {
        DefaultMQProducer producer = new DefaultMQProducer("producer_order_group_01");
        producer.setNamesrvAddr("rocketmq.jun.cn:9876");
        producer.start();

        String[] tags = new String[]{"TagA", "TagB", "TagC"};

        for (int i = 0; i < 5; i++) {
            Message message = new Message("TopicTest", tags[i % tags.length], ("Hello RocketMQ Order" + tags[i % tags.length] + "_step_" + i).getBytes(RemotingHelper.DEFAULT_CHARSET));
            message.putUserProperty("baiLi", String.valueOf(i));

            SendResult result = producer.send(message);
            System.out.println(new String(message.getBody(), RemotingHelper.DEFAULT_CHARSET));
            System.out.printf("消息发送成功 " + tags[i % tags.length] + " %s%n", result);
        }

        producer.shutdown();
    }

}
